package com.bw.app.dwd1;

import com.bw.utils.MyKafkaUtil;
import com.bw.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

/**
 * @projectName: Flinksxe
 * @fileName: DwdTradePayDetailSuc
 * @packageName: com.bw.app.dwd
 * @date: 2025/9/18 19:01
 * @author: ymw
 */
public class DwdTradePayDetailSuc {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(905));
        tenv.executeSql("create table topic_db(\n" +
                "    `database` string,\n" +
                "    `table` string,\n" +
                "    `type` string,\n" +
                "    `data` map<string,string>,\n" +
                "    `old` map<string,string>,\n" +
                "    pt as PROCTIME()\n" +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","pay_detail_suc_2212A"));
//        tenv.executeSql(MyKafkaUtil.getTopicDb("pay_detail_suc_211126"));
//                tenv.sqlQuery("select * from topic_db").execute().print();
        Table paymentInfo = tenv.sqlQuery("select\n" +
                "    `data`['user_id'] as user_id,\n" +
                "    `data`['order_id'] as order_id,\n" +
                "    `data`['payment_type'] as payment_type,\n" +
                "    `data`['callback_time'] as callback_time,\n" +
                "    `pt`\n" +
                "from topic_db\n" +
                "where  `table` = 'payment_info'\n" +
                "and `type` = 'update'\n" +
                "and `data`['payment_status'] = '1602' ");
        tenv.createTemporaryView("payment_info",paymentInfo);
//        tenv.sqlQuery("select * from payment_info").execute().print();
        tenv.executeSql("create table dwd_trade_order_detail(\n" +
                "    id String,\n" +
                "    order_id String,\n" +
                "    user_id String,\n" +
                "    sku_id String,\n" +
                "    sku_name String,\n" +
                "    sku_num String,\n" +
                "    order_price String,\n" +
                "    province_id String,\n" +
                "    activity_id String,\n" +
                "    activity_rule_id String,\n" +
                "    coupon_id String,\n" +
                "    create_time String,\n" +
                "    source_id String,\n" +
                "    source_type_id String,\n" +
                "    source_type_name String,\n" +
                "    split_activity_amount String,\n" +
                "    split_coupon_amount String,\n" +
                "    split_total_amount String,\n" +
                "    ts String,\n" +
                "    row_op_ts  TIMESTAMP_LTZ(3)\n" +
                ")"+MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail","trade_order_detail_2212"));
         tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        Table resulttable = tenv.sqlQuery("select\n" +
                "    od.id as order_detail_id,\n" +
                "    od.order_id,\n" +
                "    od.user_id,\n" +
                "    od.sku_id,\n" +
                "   CONCAT('shop_', CAST(CAST(FLOOR(RAND() * 100) + 1 AS INT) AS STRING)) AS shop_id,\n" +
                "    od.sku_name,\n" +
                "    od.province_id,\n" +
                "    od.activity_id,\n" +
                "    od.activity_rule_id,\n" +
                "    od.coupon_id,\n" +
                "    pi.payment_type as payment_type_code,\n" +
                "    b.dic_name as payment_type_name,\n" +
                "    pi.callback_time,\n" +
                "    od.source_id,\n" +
                "    od.source_type_id,\n" +
                "    od.source_type_name,\n" +
                "    od.sku_num,\n" +
                "    od.order_price,\n" +
                "    od.split_activity_amount,\n" +
                "    od.split_coupon_amount,\n" +
                "    od.split_total_amount,\n" +
                "    od.row_op_ts as row_op_ts\n" +
                "from  payment_info pi\n" +
                "join dwd_trade_order_detail od\n" +
                "on pi.order_id = od.order_id\n" +
                "join `base_dic` for system_time  as of pi.pt as b\n" +
                "on pi.payment_type = b.dic_code");
        tenv.createTemporaryView("result_table",resulttable);
//        tenv.sqlQuery("select * from result_table").execute().print();
        tenv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                "    order_detail_id String,\n" +
                "    order_id String,\n" +
                "    user_id String,\n" +
                "    sku_id String,\n" +
                "    shop_id String,\n" +
                "    sku_name String,\n" +
                "    province_id String,\n" +
                "    activity_id String,\n" +
                "    activity_rule_id String,\n" +
                "    coupon_id String,\n" +
                "    payment_type_code String,\n" +
                "    payment_type_name String,\n" +
                "    callback_time String,\n" +
                "    source_id String,\n" +
                "    source_type_id String,\n" +
                "    source_type_name String,\n" +
                "    sku_num String,\n" +
                "    order_price String,\n" +
                "    split_activity_amount String,\n" +
                "    split_coupon_amount String,\n" +
                "    split_payment_amount String,\n" +
                "    row_op_ts TIMESTAMP_LTZ(3),\n" +
                "    primary key(order_detail_id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_log_pay_detail_suc"));
        tenv.executeSql("insert into dwd_trade_pay_detail_suc select * from result_table");
        tenv.toAppendStream(resulttable, Row.class).print("====>");
        env.execute("DwdTradePayDetailSuc");
    }
}
